package kafka.examples.my;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 *
 */
public class MyConsumer {

    private static final String TOPIC = "topic2";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        //props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // consumer故障检测相关的参数
        // kafka多长时间感知不到一个consumer就认为他故障了
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        //consumer发心跳给broker的频率, 在有一个consumer故障了后, broker就会通过心跳下发rebalance的指令给其他的consumer通知他们进行rebalance的操作
        //建议不要超过session.timeout.ms的1/3.
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);

        KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));

        while(true) {
            try {
                ConsumerRecords<Integer, String> records = consumer.poll(1000);
                if (records.isEmpty()) {
                    System.out.println("Received no data");
                    Sleep.sleep(TimeUnit.SECONDS, 1);
                }
                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.println("Received[Key: " + record.key() + ", message: " + record.value() + " at offset " + record.offset());
                }
                consumer.commitSync();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

}
